package com.doitedu.core2

import com.doitedu.utils.SparkUtil
import org.apache.spark.util.LongAccumulator

/**
 * @Date: 22.7.4 
 * @Author: HANGGE
 * @qq: 598196583
 * @Tips: 学大数据 ,到多易教育
 * @Description:
 */
object C06_累加器 {
  def main(args: Array[String]): Unit = {
    val sc = SparkUtil.getSparkContext("累加器...")

    val rdd = sc.textFile("data/word.txt")

    /**
     * 使用全局计数集 统计全局结果
     *   每个分区内统计  全局汇总
     *   累加器
     */
      //    ------------------------
    val cntLines: LongAccumulator = sc.longAccumulator("cntLInes")

    // Driver变量
    var cnt = 0
    // 算子中的函数在不同的节点上执行
     rdd.foreach(line => {
      // 变量的复制   到Task所在的机器上
      cnt += 1
      //           ----------------------
      cntLines.add(1)

    })
    // 上面的cnt+=1  的cnt和当前的cnt根部不在一个进程中
   // println(cnt)
    //触发行动算子   累加器才会在各个Task上执行
  /*  rdd2.collect()
    rdd2.collect()
    rdd2.collect()*/
    // 获取累加器最后的值----------------------
    println(cntLines.value)
    sc.stop
  }

}
